﻿// AmiClient.cs
// Copyright © 2013 Alex Forster

namespace Ami
{
	using System;
	using System.Collections.Concurrent;
	using System.Collections.Generic;
	using System.Diagnostics;
	using System.Diagnostics.Contracts;
	using System.IO;
	using System.Text;
	using System.Threading;
	using System.Threading.Tasks;
	using ByteArrayExtensions;

	public sealed class AmiClient: IObservable<AmiMessage>, IDisposable
	{
		//private readonly SemaphoreSlim PublishBarrier = new SemaphoreSlim( 1, 1 );

		private Stream Stream;

		public AmiClient( Stream stream )
		{
			Contract.Requires<ArgumentNullException>( stream != null );
			
			this.Stream = stream;

			try
			{
				//this.PublishBarrier.Wait();

				var amiHelloBytes = new Byte[27]; // Asterisk Call Manager/1.1\r\n

				var amiHelloByteCount = this.Stream.Read( amiHelloBytes, 0, amiHelloBytes.Length );

				var amiHello = Encoding.ASCII.GetString( amiHelloBytes, 0, amiHelloByteCount );

				if( amiHello.StartsWith( "Asterisk Call Manager" ) == false )
				{
					throw new InvalidOperationException( "The underlying stream does not offer AMI" );
				}

				ThreadPool.QueueUserWorkItem( o => Recv() );
			}
			catch( IOException )
			{
			}
			finally
			{
				//this.PublishBarrier.Release();
			}
		}

		public event EventHandler<AmiMessageEventArgs> MessageSent;

		public event EventHandler<AmiMessageEventArgs> MessageReceived;

		private Byte[] RecvBuffer = new Byte[0];

		private void Recv()
		{
			var data = new Byte[524288];

			var dataBytes = 0;

			try
			{
				while( this.Stream != null && ( dataBytes = this.Stream.Read( data, 0, data.Length ) ) > 0 )
				{
					this.OnRecv( data.Slice( 0, dataBytes ) );
				}
			}
			catch( IOException )
			{
			}

			lock( this.Subscriptions )
			{
				foreach( var subscription in this.Subscriptions )
				{
					subscription.Completed();
				}
			}
		}

		private void OnRecv( Byte[] result )
		{
			Contract.Requires<ArgumentNullException>( result != null );

			this.RecvBuffer = this.RecvBuffer.Append( result );

			Int32 messageDelimiter;

			while( ( messageDelimiter = this.RecvBuffer.Find( new Byte[] { 0x0d, 0x0a, 0x0d, 0x0a } ) ) > -1 )
			{
				var messageBytes = this.RecvBuffer.Slice( 0, messageDelimiter + 4 );

				if( this.MessageReceived != null )
				{
					Task.Factory.StartNew( () => this.MessageReceived( this, new AmiMessageEventArgs( messageBytes ) ) );
				}

				this.DispatchMessage( messageBytes, DateTime.UtcNow );

				this.RecvBuffer = this.RecvBuffer.Slice( messageDelimiter + 4 );
			}
		}

		private void DispatchMessage( Byte[] buf, DateTime received )
		{
			Contract.Requires<ArgumentNullException>( buf != null );

			AmiException error;

			var message = AmiMessage.Deserialize( buf, out error );

			if( message == null || error != null )
			{
				lock( this.Subscriptions )
				{
					foreach( var subscription in this.Subscriptions )
					{
						subscription.Error( error );
					}
				}
			}

			AmiMessage result = null;

			var converters = new List<MessageConverter>
			{
				AmiEventMessage.Converter,
				AmiResponseMessage.Converter,
				AmiActionMessage.Converter,
			};

			for( var i = 0; i < converters.Count && result == null; i++ )
			{
				result = ( converters[i] != null ) ? converters[i].Invoke( message ) : null;
			}
			
			if( result != null )
			{
				lock( this.Subscriptions )
				{
					foreach( var subscription in this.Subscriptions )
					{
						subscription.Next( result );
					}
				}
			}
			else
			{
				lock( this.Subscriptions )
				{
					foreach( var subscription in this.Subscriptions )
					{
						subscription.Error( new AmiException( "Unhandled message" ) );
					}
				}
			}
		}

		private sealed class FutureSubscription: IObserver<AmiMessage>
		{
			private readonly ConcurrentDictionary<String, TaskCompletionSource<AmiResponseMessage>> ExpectedResponses
				= new ConcurrentDictionary<String, TaskCompletionSource<AmiResponseMessage>>();

			public TaskCompletionSource<AmiResponseMessage> Expect( String actionID )
			{
				Contract.Requires<ArgumentNullException>( actionID != null );

				return this.ExpectedResponses.GetOrAdd( actionID, new TaskCompletionSource<AmiResponseMessage>() );
			}

			public void OnNext( AmiMessage value )
			{
				var response = value as AmiResponseMessage;

				if( response == null ) return;

				TaskCompletionSource<AmiResponseMessage> completionSource;

				if( this.ExpectedResponses.TryRemove( response.ActionID, out completionSource ) )
				{
					completionSource.SetResult( response );
				}
			}

			public void OnError( Exception error )
			{
			}

			public void OnCompleted()
			{
			}
		}

		private FutureSubscription FutureResponses;

		public Task<AmiResponseMessage> Publish( AmiActionMessage message )
		{
			Contract.Requires<ArgumentNullException>( message != null );

			if( this.FutureResponses == null )
			{
				this.FutureResponses = new FutureSubscription();
				this.Subscribe( this.FutureResponses );
			}

			try
			{
				//PublishBarrier.Wait();

				var tcs = this.FutureResponses.Expect( message.ActionID );

				var messageBytes = AmiMessage.Serialize( message );

				this.Stream.Write( messageBytes, 0, messageBytes.Length );

				if( this.MessageSent != null )
				{
					Task.Factory.StartNew( () => this.MessageSent( this, new AmiMessageEventArgs( messageBytes ) ) );
				}

				return tcs.Task;
			}
			finally
			{
				//PublishBarrier.Release();
			}
		}

		public IDisposable Subscribe( IObserver<AmiMessage> observer, Queue<AmiMessage> preambleMssageQueue )
		{
			Contract.Requires<ArgumentNullException>( observer != null );

			Contract.Assert( this.Stream != null );

			lock( this.Subscriptions )
			{
				var subscription = this.Subscriptions.Find( sub => sub.Observer.Equals( observer ) );

				if( subscription != null )
				{
					return subscription;
				}

				subscription = new ClientSubscription( this, observer, preambleMssageQueue );

				this.Subscriptions.Add( subscription );

				return subscription;
			}
		}

		public IDisposable Subscribe( IObserver<AmiMessage> observer )
		{
			return this.Subscribe( observer, new Queue<AmiMessage>() );
		}

		private readonly List<ClientSubscription> Subscriptions = new List<ClientSubscription>();

		private sealed class ClientSubscription: IDisposable
		{
			private ConcurrentQueue<Tuple<Action, DateTime>> EventQueue = new ConcurrentQueue<Tuple<Action, DateTime>>();

			private readonly Thread EventQueueProcessor;

			public ClientSubscription( AmiClient client, IObserver<AmiMessage> observer, Queue<AmiMessage> preambleMessageQueue )
			{
				Contract.Requires<ArgumentNullException>( client != null );

				Contract.Requires<ArgumentNullException>( observer != null );

				this.Client = client;

				this.Observer = observer;

				while( preambleMessageQueue.Count > 0 )
				{
					var msg = preambleMessageQueue.Dequeue();
					if( this.EventQueue != null ) this.EventQueue.Enqueue( new Tuple<Action, DateTime>( () => this.Observer.OnNext( msg ), DateTime.UtcNow ) );
				}

				this.EventQueueProcessor = new Thread( this.ProcessEventQueue )
				{
					Name = "AmiClient",
					IsBackground = false,
				};

				this.EventQueueProcessor.Start();
			}

			private void ProcessEventQueue()
			{
				try
				{
					while( this.EventQueue != null )
					{
						Tuple<Action, DateTime> entry;
						if( this.EventQueue.TryDequeue( out entry ) == false )
						{
							Thread.Sleep( 1 );
							continue;
						}

						Task.Factory.StartNew( entry.Item1, TaskCreationOptions.PreferFairness );
					}
				}
				catch( ThreadAbortException )
				{
				}
			}

			public AmiClient Client
			{
				get;
				private set;
			}

			public IObserver<AmiMessage> Observer
			{
				get;
				private set;
			}

			internal void Next( AmiMessage value )
			{
				if( this.EventQueue != null ) this.EventQueue.Enqueue( new Tuple<Action, DateTime>( () => this.Observer.OnNext( value ), DateTime.UtcNow ) );
			}

			internal void Error( Exception exception )
			{
				if( this.EventQueue != null ) this.EventQueue.Enqueue( new Tuple<Action, DateTime>( () => this.Observer.OnError( exception ), DateTime.UtcNow ) );
			}

			internal void Completed()
			{
				if( this.EventQueue != null ) this.EventQueue.Enqueue( new Tuple<Action, DateTime>( () => this.Observer.OnCompleted(), DateTime.UtcNow ) );
			}

			public void Dispose()
			{
				lock( this.Client.Subscriptions )
				{
					var idxOfThis = this.Client.Subscriptions.IndexOf( this );

					if( idxOfThis > -1 )
					{
						this.Client.Subscriptions.RemoveAt( idxOfThis );
					}
				}

				this.EventQueue = null;

				this.EventQueueProcessor.Join( 1000 );
			}
		}

		public void Dispose()
		{
			if( this.Stream == null )
			{
				throw new ObjectDisposedException( "this" );
			}

			this.Stream = null;

			lock( this.Subscriptions )
			{
				foreach( var s in this.Subscriptions )
				{
					s.Completed();
				}

				this.Subscriptions.Clear();
			}
		}
	}
}
